package cn.edu.flink.scala.tutorial.watermark

import cn.edu.flink.scala.tutorial.source.Event
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkGenerator, WatermarkGeneratorSupplier, WatermarkOutput, WatermarkStrategy}
import org.apache.flink.streaming.api.scala._

import java.time.Duration
object WatermarkStrategyTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)
    env.getConfig.setAutoWatermarkInterval(500L)  // 周期性水位线生成，默认200L

    val events = List(
      Event("Mary", "home/", 10000L),
      Event("Mary", "cart/", 20000L))

    val eventsDS = env.fromCollection(events)

    val serializableTimestampAssigner = new SerializableTimestampAssigner[Event] {
      override def extractTimestamp(t: Event, l: Long): Long = t.timestamp
    }


    // 有序数据流
    eventsDS.assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps()
      .withTimestampAssigner(serializableTimestampAssigner))

    // 乱序数据流
    eventsDS.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(2L))
    .withTimestampAssigner(serializableTimestampAssigner))


    // 自定义实现水位线策略
    eventsDS.assignTimestampsAndWatermarks(new WatermarkStrategy[Event] {
      override def createWatermarkGenerator(context: WatermarkGeneratorSupplier.Context): WatermarkGenerator[Event] = {
        new WatermarkGenerator[Event] {
          override def onEvent(t: Event, l: Long, watermarkOutput: WatermarkOutput): Unit = ???  // 对收到的数据处理

          override def onPeriodicEmit(watermarkOutput: WatermarkOutput): Unit = ???  // 周期调用，提交水位线
        }
      }
    }.withTimestampAssigner(serializableTimestampAssigner))  // 也可以在WatermarkStrategy重写createTimestampAssigner方法实现时间提取
  }
}
